package com.alibaba.nacos.naming.core.v2.client.manager.impl;

import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.core.remote.ClientConnectionEventListener;
import com.alibaba.nacos.core.remote.Connection;
import com.alibaba.nacos.naming.core.v2.client.Client;
import com.alibaba.nacos.naming.core.v2.client.ClientAttributes;
import com.alibaba.nacos.naming.core.v2.client.factory.impl.ConnectionBasedClientFactory;
import com.alibaba.nacos.naming.core.v2.client.impl.ConnectionBasedClient;
import com.alibaba.nacos.naming.core.v2.client.manager.ClientManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

import static com.alibaba.nacos.naming.core.v2.event.client.ClientEvent.ClientDisconnectEvent;
import static com.alibaba.nacos.naming.core.v2.event.client.ClientOperationEvent.ClientReleaseEvent;

@Component
@Slf4j
public class ConnectionBasedClientManager
        extends ClientConnectionEventListener
        implements ClientManager {

    @PostConstruct
    public void init() {
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(r -> Thread.ofVirtual().unstarted(r));
        scheduledExecutorService.scheduleWithFixedDelay(new ExpiredClientCleaner(), 0, 100, TimeUnit.MILLISECONDS);
    }

    @Autowired
    private ConnectionBasedClientFactory connectionBasedClientFactory;

    private final ConcurrentHashMap<String, ConnectionBasedClient> clients = new ConcurrentHashMap<>() {
        @Override
        public ConnectionBasedClient remove(Object key) {
            log.warn("remove client key={} size={}", key, size() - 1);
            return super.remove(key);
        }

        @Override
        public ConnectionBasedClient computeIfAbsent(String key,
                                                     Function<? super String, ? extends ConnectionBasedClient> mappingFunction) {
            log.info("compute client key={} size={}", key, size() + 1);
            return super.computeIfAbsent(key, mappingFunction);
        }
    };

    @Override
    public Client getClient(String clientId) {
        return clients.get(clientId);
    }

    @Override
    public void clientConnected(Connection connect) {
        if (!Objects.equals(MapUtils.getString(connect.getMetaInfo().getLabels(), "module"), "naming")) {
            log.warn("仅支持module = naming");
            return;
        }

        ClientAttributes clientAttributes = new ClientAttributes();
        clientAttributes.addClientAttribute("connectionType", connect.getMetaInfo().getConnectType());
        clientAttributes.addClientAttribute("connectionMetadata", connect.getMetaInfo());
        String connectionId = connect.getMetaInfo().getConnectionId();

        ConnectionBasedClient client = connectionBasedClientFactory.newClient(connectionId, clientAttributes);
        clientConnected(client);
    }

    @Override
    public void clientDisConnected(Connection connect) {
        clientDisconnected(connect.getMetaInfo().getConnectionId());
    }

    public boolean clientConnected(final Client client) {
        clients.computeIfAbsent(client.getClientId(), key -> {
            log.info("Client connection id:{} have:{}", client.getClientId(), clients.size());
            return (ConnectionBasedClient) client;
        });

        return true;
    }

    public boolean clientDisconnected(String clientId) {
        ConnectionBasedClient removed = clients.remove(clientId);
        if (removed == null) {
            return true;
        }

        boolean responsibleClient = isResponsibleClient(removed);
        NotifyCenter.publishEvent(new ClientReleaseEvent(removed, responsibleClient));
        NotifyCenter.publishEvent(new ClientDisconnectEvent(removed, responsibleClient));
        return true;
    }

    @Override
    public boolean isResponsibleClient(Client client) {
        if (client instanceof ConnectionBasedClient connectionBasedClient) {
            return connectionBasedClient.isNative();
        }
        return false;
    }

    @Override
    public void syncClientConnected(String clientId,
                                    ClientAttributes attributes) {
        ConnectionBasedClient client = connectionBasedClientFactory.newSyncedClient(clientId, attributes);
        clientConnected(client);
    }

    private class ExpiredClientCleaner
            implements Runnable {
        @Override
        public void run() {
            long currentTimeMillis = System.currentTimeMillis();
            for (String clientId : clients.keySet()) {
                Client client = getClient(clientId);
                if (client == null && client.isExpire(currentTimeMillis)) {
                    clientDisconnected(clientId);
                }
            }
        }
    }
}
